package org.uiot.process

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.typeinfo.TypeHint
import org.apache.flink.api.java.tuple.Tuple1
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.util.Collector
import org.junit.Test

class ExampleTests extends CodecTestBase {
    @Test
    void testing() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        env.parallelism = 1
        IntegerCollectSink.values.clear()

        when:
        env.fromElements(1)
                .addSink(new IntegerCollectSink())

        and:
        env.execute()

        then:
        assert IntegerCollectSink.values.first()
    }

    // create a testing sink
    private static class IntegerCollectSink implements SinkFunction<Integer> {

        // must be static
        public static final List<Integer> values = []

        @Override
        synchronized void invoke(Integer value) throws Exception {
            values.add(value)
        }
    }

    static class MyPojo {
        private String val

        MyPojo(String v) {
            val = v
        }
    }

    @Test
    void testingPojo() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        env.parallelism = 1
        MyPojoCollectSink.values.clear()

        when:
        env.fromElements(new MyPojo("aaa"))
                .addSink(new MyPojoCollectSink())

        and:
        env.execute()

        then:
        assert MyPojoCollectSink.values.first().val == "aaa"
    }

    // create a testing sink
    private static class MyPojoCollectSink implements SinkFunction<MyPojo> {

        // must be static
        public static final List<MyPojo> values = []

        @Override
        synchronized void invoke(MyPojo value) throws Exception {
            values.add(value)
        }
    }

    @Test
    void testingTuple1ContainsMapField() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        TupleCollectSink.values.clear()

        when:
        env.fromElements(new Tuple1(["aa": 11])).returns(new TypeHint<Tuple1<Map<String, Object>>>(){})
                .addSink(new TupleCollectSink())

        env.execute()

        then:
        assert !TupleCollectSink.values.isEmpty()
        assert TupleCollectSink.values.first()
        def tuple = TupleCollectSink.values.first() as Map
        assert tuple.aa == 11
        println TupleCollectSink.values.first()
    }

    @Test
    void testingFlatMapConvertIntegerToMapField() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        env.parallelism = 1
        TupleCollectSink.values.clear()

        when:
        env.fromElements(new Tuple1(1), new Tuple1(10))
                .flatMap(new MyFlatMap())
                .addSink(new TupleCollectSink())

        env.execute()

        then:
        assert !TupleCollectSink.values.isEmpty()
        assert TupleCollectSink.values.first()
        println TupleCollectSink.values.first()
    }

    static class MyFlatMap implements FlatMapFunction<Tuple1<Integer>, Tuple1<Map<String, ?>>> {
        @Override
        void flatMap(Tuple1<Integer> value, Collector<Tuple1<Map<String, ?>>> out) throws Exception {
            out.collect(new Tuple1<>(["val": value.f0]))
        }
    }
    // create a testing sink
    private static class TupleCollectSink implements SinkFunction<Tuple1<Map<String, ?>>> {

        // must be static
        public static final List<Tuple1<Map<String, ?>>> values = []

        @Override
        synchronized void invoke(Tuple1<Map<String, ?>> value) throws Exception {
            values.add(value.f0)
        }
    }

    @Test
    void testingTupleContainsByteArrayField() {
        def env = StreamExecutionEnvironment.getExecutionEnvironment()

        env.fromElements(new Tuple1("abc".bytes))
                .addSink(new BytesTupleCollectSink())

        env.execute()

        assert !BytesTupleCollectSink.values.isEmpty()

        assert new String(BytesTupleCollectSink.values.first().f0) == "abc"
    }
    // create a testing sink
    private static class BytesTupleCollectSink implements SinkFunction<Tuple1<byte[]>> {

        // must be static
        public static final List<Tuple1<byte[]>> values = []

        @Override
        synchronized void invoke(Tuple1<byte[]> value) throws Exception {
            values.add(value)
        }
    }

    static class MyTuple1 extends Tuple1<byte[]> {
        MyTuple1(){
            super()
        }
        MyTuple1(byte[] first) {
            super(first)
        }
    }

    @Test
    void testingNamingTupleContainsByteArrayField() {
        def env = StreamExecutionEnvironment.getExecutionEnvironment()

        env.fromElements(new MyTuple1("abc".bytes))
                .addSink(new BytesTupleCollectSink2())

        env.execute()

        assert !BytesTupleCollectSink.values.isEmpty()

        assert new String(BytesTupleCollectSink.values.first().f0) == "abc"
    }

    // create a testing sink
    private static class BytesTupleCollectSink2 implements SinkFunction<MyTuple1> {

        // must be static
        public static final List<MyTuple1> values = []

        @Override
        synchronized void invoke(MyTuple1 value) throws Exception {
            values.add(value)
        }
    }

    @Test
    void testTuple2() {
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        Tuple2CollectSink.values.clear()

        env.fromElements(new Tuple2<String, Integer>("12345", 1))
                .addSink(new Tuple2CollectSink())

        env.execute()

        assert !Tuple2CollectSink.values.isEmpty()
        assert Tuple2CollectSink.values.first()
    }

    // create a testing sink
    private static class Tuple2CollectSink implements SinkFunction<Tuple2<String, Integer>> {

        // must be static
        public static final List<Tuple2<String, Integer>> values = []

        @Override
        synchronized void invoke(Tuple2<String, Integer> value) throws Exception {
            values.add(value)
        }
    }

    @Test
    void testTupleContainedGroovyScriptString() {
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        IntegerTupleCollectSink.values.clear()

        env.fromElements(new Tuple2<String, Integer>("x + 1", 1))
                .flatMap(new GroovyScriptFlatMapFunc())
                .addSink(new IntegerTupleCollectSink())

        env.execute()

        assert !IntegerTupleCollectSink.values.isEmpty()
        assert IntegerTupleCollectSink.values.first().f0 == 2
    }

    @Test(expected = org.apache.flink.runtime.client.JobExecutionException.class)
    void testTupleContainedGroovyScriptFailure() {
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        IntegerTupleCollectSink.values.clear()

        Class<Script> scriptClass = new GroovyClassLoader().parseClass("x + 1")

        env.fromElements(new Tuple2<String, Integer>(scriptClass, 1))
                .flatMap(new GroovyScriptFlatMapFunc2())
                .print()

        env.execute()
    }

    static class GroovyScriptFlatMapFunc implements FlatMapFunction<Tuple2<String, Integer>, Tuple1<Integer>>, Serializable {
        @Override
        void flatMap(Tuple2<String, Integer> value, Collector<Tuple1<Integer>> out) throws Exception {
            Class<Script> scriptClass = new GroovyClassLoader().parseClass(value.f0)
            def script = scriptClass.newInstance() as Script
            script.binding.setVariable("x", value.f1)
            out.collect(new Tuple1<Integer>(script.run() as Integer))
        }
    }

    static class GroovyScriptFlatMapFunc2 implements FlatMapFunction<Tuple2<Class<Script>, Integer>, Tuple1<Integer>>, Serializable {
        @Override
        void flatMap(Tuple2<Class<Script>, Integer> value, Collector<Tuple1<Integer>> out) throws Exception {
            def script = value.f0.newInstance() as Script
            script.binding.setVariable("x", value.f1)
            out.collect(new Tuple1<Integer>(script.run() as Integer))
        }
    }

    // create a testing sink
    private static class IntegerTupleCollectSink implements SinkFunction<Tuple1<Integer>> {

        // must be static
        public static final List<Tuple1<Integer>> values = []

        @Override
        synchronized void invoke(Tuple1<Integer> value) throws Exception {
            values.add(value)
        }
    }
}
